package com.gitee.sparta.core.process.operate.redis;

import cn.hutool.core.util.StrUtil;
import com.gitee.sparta.common.SpartaConstant;
import com.gitee.sparta.common.SpartaErrorEnum;
import com.gitee.sparta.common.SpartaLuaConstant;
import com.gitee.sparta.common.exception.Assert;
import com.gitee.sparta.common.exception.SpartaException;
import com.gitee.sparta.common.helpers.*;
import com.gitee.sparta.core.bean.metadata.Metadata;
import com.gitee.sparta.core.process.feature.SpartaTaskDispatcherFeature;
import com.gitee.sparta.core.process.feature.SpartaWorkerFeature;
import com.gitee.sparta.core.process.operate.AbstractTimeoutTabOperate;
import com.gitee.sparta.core.property.SpartaCoreProperties;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RScript;
import org.redisson.api.RedissonClient;
import org.redisson.client.codec.StringCodec;

import java.time.Duration;
import java.util.*;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.stream.Collectors;

/**
 * lua 对 timeout_tab 的实现
 *
 * @param <V>
 * @author zyred
 * @since v 0.0.1
 */
@Slf4j
public class LuaAbstractTimeoutTabOperate<V> extends AbstractTimeoutTabOperate<V> {
    private static volatile boolean enable = true;

    /**
     * 异步请求，放入到此队列
     **/
    private final LinkedBlockingQueue<Metadata<V>> submittedRequests = new LinkedBlockingQueue<>();

    private final RedissonClient client;
    private final SpartaCoreProperties properties;

    public LuaAbstractTimeoutTabOperate(SpartaCoreProperties spartaCoreProperties, RedissonClient redissonClient) {
        this.client = redissonClient;
        this.properties = spartaCoreProperties;
        super.setName(SpartaConstant.SPARTA_WRIT_THREAD_NAME);
        super.setDaemon(true);
        super.start();
    }

    @Override
    public void writeAsync(String metaId, String topic, V body, long expire, boolean broadcast) {
        this.canWrite(body, topic, broadcast);

        Assert.lessEq(expire, SpartaErrorEnum.NUMBER_OF_LESS_EQ_ZERO);
        Metadata<V> normalMetadata = this.createMetadata(metaId, topic, body, 0, Collections.singleton(expire), broadcast);
        this.submittedRequests.add(normalMetadata);
    }

    @Override
    public void writeAsync(String metaId, String topic, V body, List<Duration> times, boolean broadcast) {
        this.canWrite(body, topic, broadcast);

        Assert.isEmpty(times, SpartaErrorEnum.SEGMENT_NOT_EMPTY);
        final int segmentSize = times.size();
        Set<Long> millis = DurationHelper.parseMillis(times);
        Metadata<V> metadata = this.createMetadata(metaId, topic, body, segmentSize, millis, broadcast);
        this.submittedRequests.add(metadata);
    }

    @Override
    public void write(String metaId, String topic, V body, long expire, boolean broadcast) {
        this.canWrite(body, topic, broadcast);

        Assert.lessEq(expire, SpartaErrorEnum.NUMBER_OF_LESS_EQ_ZERO);
        Metadata<V> metadata = this.createMetadata(metaId, topic, body, 0, Collections.singleton(expire), broadcast);

        this.doWrite(metadata);
    }

    @Override
    public void write(String metaId, String topic, V body, List<Duration> times, boolean broadcast) {
        this.canWrite(body, topic, broadcast);

        Assert.isEmpty(times, SpartaErrorEnum.SEGMENT_NOT_EMPTY);
        Set<Long> millis = DurationHelper.parseMillis(times);
        Metadata<V> metadata = this.createMetadata(metaId, topic, body, millis.size(), millis, broadcast);
        this.doWrite(metadata);
    }

    @SuppressWarnings("all")
    @Override
    public List<Object> carry(String timeoutTabKey, String readableTabKey, long currentTimestamp, String balance) {

        Map<String, String> cache = SpartaLuaConstant.getSha(SpartaConstant.CARRY);
        final List<Object> keys = Arrays.asList(readableTabKey, timeoutTabKey);

        RScript script = this.client.getScript(CodecCache.getCodec(StringCodec.class));
        Object result = script.evalSha(RScript.Mode.READ_ONLY, cache.get(SpartaConstant.LUA_SHA_KEY),
                RScript.ReturnType.MULTI, keys, currentTimestamp, balance, SpartaConstant.BROADCAST_PREFIX);

        if (Objects.nonNull(result) && result instanceof List) {
            return (List<Object>) result;
        }
        return Collections.emptyList();
    }

    @Override
    public void shutdown() {
        // 不允许添加
        enable = false;
        // 大于 0 的情况， 说明还有值，这里需要帮助 #run() 方法执行搬运
        if (!this.submittedRequests.isEmpty()) {
            Metadata<V> metadata = this.submittedRequests.poll();
            if (Objects.nonNull(metadata)) {
                this.doWrite(metadata);
            }
        }
    }

    /**
     * 执行写入 redis 异步提交到 submittedRequests 阻塞队列的任务
     * 将准备完毕的 metadata 写入到 redis 的 metadat_tab 与 timeout_tab 中
     */
    @SuppressWarnings("all")
    @Override
    public void run() {
        while (enable) {
            try {
                // 阻塞队列，如果没有值的情况，需要进入阻塞
                Metadata<V> metadata = this.submittedRequests.take();
                if (Objects.isNull(metadata)) {
                    if (log.isDebugEnabled()) {
                        log.debug("[Sparta] Read queue content is empty.");
                    }
                    continue;
                }
                this.doWrite(metadata);
            } catch (InterruptedException ex) {
                log.error("[Sparta] Async write exception ", ex.getMessage(), ex);
            }
        }
        log.warn("[Sparta] Async write thread destroy {} closed", currentThread().getName());
    }


    @Override
    public void closeRollback() {
        final String isolation = this.properties.getIsolation();
        final String systemName = this.properties.getSystemName();
        final String timeoutTabKey = RedisKeyHelper.tabKey(systemName, SpartaConstant.TABLE_TIMEOUT_NAME);

        // 将所有 readable_tab key 组装起来，提供给 lua 脚本遍历
        final String readableTabKeys = SpartaTaskDispatcherFeature.getReadableTabKeys(systemName, isolation);
        if (StrUtil.isBlank(readableTabKeys)) {
            log.warn("[Sparta] Close rollback, notfound project listen topic ..");
            return;
        }

        // 开始执行逻辑
        Map<String, String> cache = SpartaLuaConstant.getSha(SpartaConstant.SHUTDOWN_AFTER_ROLLBACK);
        final List<Object> keys = Collections.singletonList(timeoutTabKey);

        try {
            final long millis = System.currentTimeMillis();
            RScript script = this.client.getScript(CodecCache.getCodec(StringCodec.class));
            script.evalShaAsync(RScript.Mode.READ_ONLY, cache.get(SpartaConstant.LUA_SHA_KEY),
                    RScript.ReturnType.VALUE, keys, millis, readableTabKeys);
        } catch (Exception ex) {
            log.error("[Sparta] Close rollback exception ：{}", ex.getMessage(), ex);
            ScriptShaHelper.reloadScriptSha(ex, this.client, properties);
        }
    }

    /**
     * 开始真正往 redis 中写入数据
     *
     * @param metadata 被设置好的内容
     */
    private void doWrite(Metadata<V> metadata) {
        final String systemName = this.properties.getSystemName();
        final String metadataTabKey = RedisKeyHelper.tabKey(systemName, SpartaConstant.TABLE_METADATA_NAME);
        final String timeoutTabKey = RedisKeyHelper.tabKey(systemName, SpartaConstant.TABLE_TIMEOUT_NAME);
        final List<Object> keys = Arrays.asList(metadataTabKey, timeoutTabKey);

        String fullTopic = RedisKeyHelper.fieldName(metadata.getTopic(), metadata.getMetaId());

        // 启用数据压缩
        String metadataJson = this.properties.isEnableMetadataCompress()
                ? ZipStrHelper.compress(JacksonHelper.serialize(metadata))
                : JacksonHelper.serialize(metadata);

        Map<String, String> cache;
        RScript script = this.client.getScript(CodecCache.getCodec(StringCodec.class));

        try {
            // 写分段消息
            if (metadata.isSegment()) {
                String segment = metadata.getSegments().stream().map(String::valueOf)
                        .collect(Collectors.joining(SpartaConstant.COMMA));

                cache = SpartaLuaConstant.getSha(SpartaConstant.WRITE_SEGMENT);
                script.evalSha(
                        RScript.Mode.READ_ONLY, cache.get(SpartaConstant.LUA_SHA_KEY), RScript.ReturnType.VALUE,
                        keys, fullTopic, metadataJson, segment, SpartaConstant.LINE
                );
            }
            // 写广播
            else if (metadata.isBroadcast()) {
                cache = SpartaLuaConstant.getSha(SpartaConstant.WRITE_BROADCAST);
                String readableField = SpartaConstant.BROADCAST_PREFIX.concat(fullTopic);
                script.evalSha(
                        RScript.Mode.READ_ONLY, cache.get(SpartaConstant.LUA_SHA_KEY), RScript.ReturnType.VALUE,
                        keys, fullTopic, readableField, metadata.getExpire(), metadataJson
                );
            }
            // 普通写
            else {
                cache = SpartaLuaConstant.getSha(SpartaConstant.WRITE_NORMAL);
                script.evalSha(
                        RScript.Mode.READ_ONLY, cache.get(SpartaConstant.LUA_SHA_KEY), RScript.ReturnType.VALUE,
                        keys, fullTopic, metadata.getExpire(), metadataJson
                );
            }
            SpartaWorkerFeature.interruptWorker();
        } catch (Exception ex) {
            log.error("[Sparta] write exception, metadata: {}, exception : {} ", metadataJson, ex.getMessage(), ex);
            ScriptShaHelper.reloadScriptSha(ex, this.client, properties);
        }
    }

    /**
     * 创建 metadata 对象
     *
     * @param metaId      自定义的消息ID
     * @param topic       topic
     * @param body        消息类型，泛型任意
     * @param segmentSize 批量消息的标识
     * @param times       时间集合，如果 segmentSize > 0 的时候，times 一定不为 null
     * @param broadcast   是否是广播，true: 广播，false：非广播
     * @return Metadata 对象
     */
    private Metadata<V> createMetadata(String metaId, String topic, V body, int segmentSize, Set<Long> times, boolean broadcast) {
        Assert.len(metaId, SpartaConstant.META_ID_MIN_LEN, SpartaConstant.META_ID_MAX_LEN);
        Assert.isEmpty(times, SpartaErrorEnum.SEGMENT_NOT_EMPTY);
        Assert.isTrue(!RedisKeyHelper.requirements(topic), SpartaErrorEnum.TOPIC_NOT_REQUIREMENTS);

        Metadata<V> metadata = new Metadata<>();

        metadata.setBody(body);
        metadata.setTopic(topic);
        metadata.setSegmentSize(segmentSize);
        metadata.setMetaId(StrUtil.isBlank(metaId) ? MetaIdHelper.nextId() : metaId);
        metadata.setBroadcast(broadcast);

        if (segmentSize > 0) {
            metadata.setExpire(0L);
            metadata.setSegments(times);
        } else {
            metadata.setSegments(Collections.emptySet());
            Long expire = new ArrayList<>(times).get(0);
            metadata.setExpire(System.currentTimeMillis() + expire);
        }
        return metadata;
    }

    /**
     * 检查是否满足条件，满足方可写入
     *
     * @param body      消息体，做异常打印
     * @param topic     topic
     * @param broadcast 是否是广播 true 是广播 false: 不是广播
     */
    private void canWrite(V body, String topic, boolean broadcast) {
        if (!enable) {
            throw new SpartaException("springboot 项目已经关闭，请勿写入消息: " + JacksonHelper.serialize(body));
        }
        boolean exist;
        if (broadcast) {
            // 关闭了广播功能
            if (!this.properties.isEnableBroadcast()) {
                throw new SpartaException("广播功能已经关闭，请启用广播功能后再使用：spring.sparta.enable-broadcast=true");
            }

            exist = SpartaTaskDispatcherFeature.channelExistListener(topic);
        } else {
            exist = SpartaTaskDispatcherFeature.topicExistListeners(topic);
        }
        // 检查 topic 是否有对应的监听器
        if (!exist) {
            throw new SpartaException("[" + topic + "] 不存在监听器，无法写入到队列中，请设置监听器.");
        }
    }

}
